package model

import (
	"context"
	"errors"
	commonutils "fastdp-ops/pkg/common/utils"
	. "fastdp-ops/pkg/server/api/middleware"
	"fastdp-ops/pkg/server/api/models"
	. "fastdp-ops/pkg/server/config"
	. "fastdp-ops/pkg/server/grpc"
	"fastdp-ops/pkg/server/utils" // 引入utils包
	pb "fastdp-ops/proto/exec"
	"fmt"
	"gopkg.in/yaml.v3"
	"io"
	"os"
	"path"
	"strconv"
	"sync"
	"time"
)

// Playbook 定义YAML编排文件的结构
type Playbook struct {
	Name       string  `yaml:"name"`
	Version    string  `yaml:"version"`
	ConfigFile string  `mapstructure:"config_file"`
	Stages     []Stage `yaml:"stages"` // 任务阶段列表
}

// Stage 定义一个任务阶段
type Stage struct {
	Stage string `yaml:"stage"` // 阶段名称
	Role  string `yaml:"role"`
	Tasks string `yaml:"tasks"` // 阶段包含的任务
}

var taskList []Task
var handler map[string]Task = make(map[string]Task)

// Task 定义具体任务
// Task 定义具体任务
type Task struct {
	Name    string            `yaml:"name"`
	TaskId  string            `yaml:"task_id"`
	Module  string            `yaml:"module"`
	Params  map[string]string `yaml:"params"` // 直接是 map
	Loop    []string          `yaml:"loop,omitempty"`
	When    string            `yaml:"when,omitempty"`
	Hooks   string            `yaml:"hooks,omitempty"`
	Order   string            `yaml:"order,omitempty"`
	Timeout int               `yaml:"timeout,omitempty"` // 新增：任务级超时设置
}

func PlayStage(stage Stage, verbose bool, role map[string][]string, globalConfig map[string]interface{}, workdir string) error {
	tasksDir := path.Join(workdir, stage.Tasks)
	tasksFile := path.Join(tasksDir, "tasks.yaml") // 用path.Join更安全，避免路径拼接问题
	handlersFile := path.Join(tasksDir, "handlers.yaml")

	taskList, err := parseYAMLFile(tasksFile)
	if err != nil {
		return fmt.Errorf("解析任务文件 %s 失败: %w", tasksFile, err)
	}

	// 解析处理器文件（忽略文件不存在错误）
	handlersSlice, err := parseYAMLFile(handlersFile)
	if os.IsNotExist(err) {
		// 文件不存在是正常情况，清空处理器列表
		handlersSlice = []Task{}
	} else if err != nil {
		return fmt.Errorf("解析处理器文件 %s 失败: %w", handlersFile, err)
	}
	handler = make(map[string]Task) // 重置处理器映射
	for _, task := range handlersSlice {
		if task.Name == "" {
			return fmt.Errorf("处理器文件中存在名称为空的任务")
		}
		if _, exists := handler[task.Name]; exists {
			return fmt.Errorf("处理器文件中存在重复的任务名称: %s", task.Name)
		}
		handler[task.Name] = task
	}

	// 1. 阶段元信息展示
	fmt.Printf("📌 阶段名称: %s | 任务数: %d\n", stage.Stage, len(taskList))
	err = commonutils.SyncMachineEnv()
	if err != nil {
		fmt.Println(err)
	}
	// 获取可用机器列表
	machineList := []models.Machines{}
	if err := DB.Find(&machineList).Error; err != nil {
		errMsg := "查询数据库机器列表失败"
		return fmt.Errorf("%s: %w", errMsg, err) // 返回错误，终止当前阶段
	}
	// 筛选在线主机
	onlineHosts := make(map[string]models.Machines)
	var onlineHostAddrs []string
	for _, machine := range machineList {

		if machine.Status == 1 {
			onlineHosts[machine.MachineIP] = machine
			onlineHostAddrs = append(onlineHostAddrs, machine.MachineIP)
		}
	}

	if len(onlineHosts) == 0 {
		return fmt.Errorf("无可用目标主机（未在线或未注册）")
	}
	// 打印在线主机IP
	fmt.Println("\n📌 在线主机列表：")
	for i, ip := range onlineHostAddrs {
		fmt.Printf("   %d. %s (主机名: %s)\n", i+1, ip, onlineHosts[ip].Hostname)
	}
	// 获取Agent端口
	agentPort := V.GetString("fastdp-agent.port")
	if agentPort == "" {
		agentPort = "8700" // 默认端口，防止配置缺失
	}
	if err := PreCheck(onlineHosts, agentPort); err != nil {
		fmt.Printf("❌ 部署前检查失败：%v\n", err)
		return err // 检查失败终止部署
	}
	fmt.Println("✅ 所有部署前检查通过")
	// 取目标主机与在线主机交集

	targetHosts := make([]string, 0)
	handlerHosts := make([]string, 0)
	if stage.Role == "all" {
		targetHosts = onlineHostAddrs
	}
	for _, stageHost := range role[stage.Role] {
		if _, isOnline := onlineHosts[stageHost]; isOnline {
			targetHosts = append(targetHosts, stageHost)
		} else {
			return fmt.Errorf("❌ 角色%s中目标主机%s不在线，退出执行", stage.Role, stageHost)
		}
	}

	fmt.Printf("🔍 主机筛选: 匹配成功%d台目标机器\n", len(targetHosts))
	if len(targetHosts) == 0 {
		return fmt.Errorf("无可用目标主机（未在线或未注册）")
	}
	if verbose {
		fmt.Printf("   匹配主机列表: %v\n", targetHosts) // 详细模式展示具体主机
	}

	// 执行阶段内所有任务（带任务级统计）
	var totalTasks, successTasks int
	for taskIdx, task := range taskList {
		totalTasks++
		taskName := task.Name
		if taskName == "" {
			taskName = fmt.Sprintf("未命名任务%d", taskIdx+1)
		}

		fmt.Printf("\n----- 任务 %d/%d: %s（模块: %s）-----\n",
			taskIdx+1, len(taskList), taskName, task.Module)

		// 执行任务
		allResponses, err := PlayTaskExec(task, onlineHosts, targetHosts, role, globalConfig, workdir, stage.Tasks)
		if err != nil {
			return fmt.Errorf("任务执行失败: %w", err) // 任务失败终止阶段
		}

		// 解析任务响应（按主机分组展示）
		taskSuccess := 0
		for _, resp := range allResponses {
			hostIP := resp.MachineId

			if !resp.Success {
				// 失败：红底标记+错误详情
				fmt.Printf("   ❌ 主机 %s 执行失败: %s\n", hostIP, resp.Error.Message)
			} else {
				taskSuccess++
				handlerHosts = append([]string{}, hostIP)
				if resp.Change {
					// 成功且有变更：绿底标记+变更详情
					fmt.Printf("   ✅ 主机 %s 执行成功:\n%s\n", hostIP, resp.ChangedDetails)
					if task.Hooks != "" {
						handlerResponses, err := PlayTaskExec(handler[task.Hooks], onlineHosts, handlerHosts, role, globalConfig, workdir, stage.Tasks)

						if err != nil {
							return fmt.Errorf("钩子任务执行失败: %w", err)
						}

						for _, hresp := range handlerResponses {
							hhostIP := hresp.MachineId
							if !hresp.Success {
								// 失败：红底标记+错误详情
								fmt.Printf("   ❌ 主机 %s 钩子%s执行失败: %s\n", hhostIP, task.Hooks, hresp.Error.Message)
								return fmt.Errorf("钩子任务 %s 在主机 %s 执行失败", task.Hooks, hhostIP)
							} else {
								fmt.Printf("   ✅ 主机 %s 钩子%s执行成功: %s\n", hhostIP, task.Hooks, hresp.ChangedDetails)
							}
						}

					}
				} else {
					// 成功无变更：蓝底标记
					fmt.Printf("   ℹ️ 主机 %s 执行成功:\n%s\n", hostIP, resp.ChangedDetails)
				}
				// 详细模式展示标准输出
				if verbose && resp.Stdout != "" {
					fmt.Printf("      输出: %s\n%s", resp.Stdout, resp.Stderr)
				}
			}
		}

		// 任务执行统计
		fmt.Printf("   📊 任务结果: 总%d台 -> 成功%d台 -> 失败%d台\n",
			len(targetHosts), taskSuccess, len(targetHosts)-taskSuccess)
		if taskSuccess < len(targetHosts) {
			// 任务部分失败（可根据需求决定是否终止阶段）
			return fmt.Errorf("任务 %s 在部分主机执行失败", taskName)
		}
		successTasks++
	}
	// 阶段内任务执行总结
	fmt.Printf("\n📊 阶段任务总结: 总%d个 -> 成功%d个\n", totalTasks, successTasks)
	return nil
}
func PlayTaskExec(task Task, hostSet map[string]models.Machines, execHostList []string, role map[string][]string, globalConfig map[string]interface{}, workdir string, tasks string) ([]*pb.ExecResponse, error) {
	// 1. 验证核心参数
	if task.Module == "" {
		return nil, fmt.Errorf("❌ 必需字段缺失：module（执行模块不能为空）")
	}
	if task.TaskId == "" {
		return nil, fmt.Errorf("❌ 必需字段缺失：taskId（任务唯一标识不能为空）")
	}

	// 拷贝一份，避免修改原数据
	paramsMap := make(map[string]string, len(task.Params))
	for k, v := range task.Params {
		paramsMap[k] = v
	}

	agentPort := V.GetString("fastdp-agent.port")
	grpcTimeout := time.Duration(V.GetInt("grpc.default_timeout")) * time.Second
	if grpcTimeout == 0 {
		grpcTimeout = 60 * time.Second
	}
	if agentPort == "" {
		agentPort = "8700" // 默认端口，防止配置缺失
	}

	// 3. 准备并发执行资源
	allResponses := make([]*pb.ExecResponse, 0, len(execHostList))

	var wg sync.WaitGroup
	var mu sync.Mutex
	var globalErr error
	// 4. 打印任务执行上下文
	fmt.Printf("   开始执行模块: %s（任务ID: %s）\n", task.Module, task.TaskId)
	fmt.Printf("   目标主机数: %d | Agent端口: %s\n", len(execHostList), agentPort)
	var loopItems []interface{} // 循环列表的实际数据
	isLoopTask := false
	if len(task.Loop) > 0 {
		isLoopTask = true
		mergedVars := commonutils.MergeVariables(globalConfig, map[string]interface{}{}, role)
		loopItems, globalErr = commonutils.RenderLoopVariable(task.Loop, mergedVars)

		if globalErr != nil {
			return nil, fmt.Errorf("loop结果转换失败,请检查语法: %w", globalErr)
		}
		fmt.Printf("   检测到循环任务: 共%d个循环项\n", len(loopItems))
	}
	// 5. 并发执行各主机任务
	for _, host := range execHostList {
		wg.Add(1)
		go func(host string) {
			defer wg.Done()

			// 获取机器变量
			machineVars, err := commonutils.CollectMachineVariables(host, hostSet[host], paramsMap)
			if err != nil {
				mu.Lock()
				allResponses = append(allResponses, utils.ErrorResponse(
					&pb.Execrequest{MachineId: host, TaskId: task.TaskId},
					500, "收集主机变量失败: "+err.Error()))
				mu.Unlock()
				return
			}
			// 合并全局变量和机器变量
			mergedVars := commonutils.MergeVariables(globalConfig, machineVars, role)
			mergedVars["hostSet"] = hostSet

			req := &pb.Execrequest{
				MachineId:  host,
				Module:     task.Module,
				Parameters: paramsMap,
				TaskId:     task.TaskId,
			}
			if task.When != "" {
				renderdWhen, err := commonutils.RenderWhen(task.When, mergedVars)
				run, err := isTaskShouldRun(renderdWhen)
				if err != nil {
					mu.Lock()
					allResponses = append(allResponses, utils.ErrorResponse(req, 500, "when条件判断失败: "+err.Error()))
					mu.Unlock()
					return
				}
				if !run {
					mu.Lock()
					allResponses = append(allResponses, utils.SuccessResponseWithNoChange(req, "符合when语句未执行", renderdWhen))
					mu.Unlock()
					return
				}
			}
			if isLoopTask {
				ordered := true
				if task.Order != "" {
					ordered, _ = strconv.ParseBool(task.Order) // 转换为布尔值（忽略错误，默认true）
				}
				var batchReq []*pb.Execrequest
				for itemIdx, item := range loopItems {
					itemVars := make(map[string]interface{})
					for k, v := range mergedVars {
						itemVars[k] = v
					}
					// 关键逻辑：处理item为map的情况
					if itemMap, ok := item.(map[string]string); ok {
						itemVars["item"] = itemMap
					} else {
						// 非map类型，直接注入item变量
						itemVars["item"] = item
					}
					itemParams := make(map[string]string, len(paramsMap))
					for k, v := range paramsMap {
						itemParams[k] = v
					}

					// 仅对template模块且未指定content时，读取模板文件
					if task.Module == "template" && itemParams["content"] == "" {
						tempParams := map[string]string{"src": itemParams["src"]}
						renderedTempParams, err := commonutils.RenderParams(tempParams, itemVars)
						if err != nil {
							mu.Lock()
							allResponses = append(allResponses, utils.ErrorResponse(
								&pb.Execrequest{MachineId: host, TaskId: fmt.Sprintf("%s-loop-%d", task.TaskId, itemIdx)},
								500, fmt.Sprintf("渲染模板路径参数失败: %s", err)))
							mu.Unlock()
							continue
						}
						resolvedSrc := renderedTempParams["src"]
						// 模板路径计算（使用当前任务的工作目录）
						taskDir := path.Join(workdir, tasks) // 注意：需要将stage.Tasks通过参数传入或闭包捕获
						tplPath := path.Join(taskDir, "templates", fmt.Sprintf("%s.tpl", resolvedSrc))

						// 读取模板文件内容
						if data, err := os.ReadFile(tplPath); err == nil {
							itemParams["content"] = string(data) // 将模板内容存入content参数
						} else {
							mu.Lock()
							allResponses = append(allResponses, utils.ErrorResponse(
								&pb.Execrequest{MachineId: host, TaskId: fmt.Sprintf("%s-loop-%d", task.TaskId, itemIdx)},
								500, fmt.Sprintf("读取模板文件失败: %s", err)))
							mu.Unlock()
							continue
						}
					}

					renderParams, err := commonutils.RenderParams(itemParams, itemVars)

					if err != nil {
						mu.Lock()
						allResponses = append(allResponses, utils.ErrorResponse(
							&pb.Execrequest{MachineId: host, TaskId: fmt.Sprintf("%s-loop-%d", task.TaskId, itemIdx)},
							500, fmt.Sprintf("渲染循环项%d参数失败: %s", itemIdx, err)))
						mu.Unlock()
						continue
					}
					// 添加到批量请求
					batchReq = append(batchReq, &pb.Execrequest{
						MachineId:  host,
						Module:     task.Module,
						Parameters: renderParams,
						TaskId:     fmt.Sprintf("%s-loop-%d", task.TaskId, itemIdx),
					})
				}
				// 7. 其他模块通过gRPC执行
				addr := host + ":" + agentPort
				conn, err := GetClientConn(addr)
				if err != nil {
					mu.Lock()
					allResponses = append(allResponses, utils.ErrorResponse(req, 1, "连接失败: "+err.Error()))
					mu.Unlock()
					return
				}
				defer func() {
					if err := conn.Close(); err != nil {
						fmt.Printf("   ⚠️ 主机 %s 连接关闭失败: %v\n", host, err)
					}
				}()
				if task.Timeout > 0 {
					grpcTimeout = time.Duration(task.Timeout) * time.Second
				}

				// 8. 发送gRPC请求
				client := pb.NewExecServiceClient(conn)
				batchCtx, batchCancel := context.WithTimeout(context.Background(), grpcTimeout)
				defer batchCancel()

				stream, err := client.BatchExec(batchCtx, &pb.BatchExecrequest{
					BatchId: task.TaskId,
					Tasks:   batchReq,
					Ordered: ordered,
				})
				if err != nil {
					mu.Lock()
					allResponses = append(allResponses, utils.ErrorResponse(
						&pb.Execrequest{MachineId: host, TaskId: task.TaskId},
						500, "发起批量请求失败: "+err.Error()))
					mu.Unlock()
					return
				}
				// 接收批量响应（流式接收所有循环项结果）
				for {
					resp, err := stream.Recv()
					if err == io.EOF {
						break // 所有响应接收完成
					}
					if err != nil {
						mu.Lock()
						allResponses = append(allResponses, utils.ErrorResponse(
							&pb.Execrequest{MachineId: host, TaskId: task.TaskId},
							500, "接收批量响应失败: "+err.Error()))
						mu.Unlock()
						break
					}
					// 收集正常响应
					mu.Lock()
					allResponses = append(allResponses, resp)
					mu.Unlock()
				}

			} else {
				renderParams := make(map[string]string, len(paramsMap))
				for k, v := range paramsMap {
					renderParams[k] = v
				}

				// 处理template模块的content参数（非循环场景）
				if task.Module == "template" && renderParams["content"] == "" {
					taskDir := path.Join(workdir, tasks) // 同上，需要确保stage.Tasks可访问
					tplPath := path.Join(taskDir, "templates", fmt.Sprintf("%s.tpl", renderParams["src"]))
					if data, err := os.ReadFile(tplPath); err == nil {
						renderParams["content"] = string(data)
					} else {
						mu.Lock()
						allResponses = append(allResponses, utils.ErrorResponse(
							&pb.Execrequest{MachineId: host, TaskId: task.TaskId},
							500, "读取模板文件失败: "+err.Error()))
						mu.Unlock()
						return
					}
				}
				renderParams, err := commonutils.RenderParams(renderParams, mergedVars)
				if err != nil {
					mu.Lock()
					allResponses = append(allResponses, utils.ErrorResponse(
						&pb.Execrequest{MachineId: host, TaskId: task.TaskId},
						500, "渲染参数失败,请检查语法: "+err.Error()))
					mu.Unlock()
					return
				}
				req := &pb.Execrequest{
					MachineId:  host,
					Module:     task.Module,
					Parameters: renderParams,
					TaskId:     task.TaskId,
				}

				// 7. 其他模块通过gRPC执行
				addr := host + ":" + agentPort
				conn, err := GetClientConn(addr)
				if err != nil {
					mu.Lock()
					allResponses = append(allResponses, utils.ErrorResponse(req, 1, "连接失败: "+err.Error()))
					mu.Unlock()
					return
				}
				defer conn.Close()
				if task.Timeout > 0 {
					grpcTimeout = time.Duration(task.Timeout) * time.Second
				}
				// 8. 发送gRPC请求
				client := pb.NewExecServiceClient(conn)
				ctx, cancel := context.WithTimeout(context.Background(), grpcTimeout)
				defer cancel()

				resp, err := client.Exec(ctx, req)

				mu.Lock()
				if err != nil {
					if errors.Is(err, context.DeadlineExceeded) {
						// 确认是超时导致的失败
						allResponses = append(allResponses, utils.ErrorResponse(req, 500, "gRPC调用失败: gRPC调用超时"))
						return
					}
					// gRPC调用失败输出
					fmt.Printf("   ❌ 主机 %s gRPC调用失败: %v\n", host, err)
					allResponses = append(allResponses, utils.ErrorResponse(
						req, 500, "gRPC调用失败: "+err.Error()))
				} else {
					allResponses = append(allResponses, resp)
				}
				mu.Unlock()
			}
		}(host)
	}

	wg.Wait()
	return allResponses, nil

}

func parseYAMLFile(filePath string) ([]Task, error) {
	// 读取文件内容
	data, err := os.ReadFile(filePath)
	if err != nil {
		return nil, err
	}

	// 解析 YAML 内容
	var tasks []Task
	if err := yaml.Unmarshal(data, &tasks); err != nil {
		// 增强错误信息，显示具体行号
		if yamlErr, ok := err.(*yaml.TypeError); ok {
			return nil, fmt.Errorf("YAML解析错误: %v\n具体错误: %v", err, yamlErr.Errors)
		}
		return nil, err
	}

	return tasks, nil
}
